package com.heima.kafka.simple;

import org.apache.kafka.clients.producer.*;

import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class ProducerTest {


    public static void main(String[] args) {

        //1、指定kafka的配置信息
        //用于封装配置信息
        Properties properties = new Properties();
        //设置kafka主机的地址和端口
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.200.130:9092");
        //设置消息的序列化器(kafka存数据时存的是字节数组，我们在发的时候发的是字符串。需要指定序列化器，将字符串序列化成字节数组)
        properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        //kafka确认机制: 0：发了不管   1:leader收到就认为发送成功  all:leader follower都要收到才认为成功
        properties.setProperty(ProducerConfig.ACKS_CONFIG,"all");
        //如果采用同步发送不出去，指定重试次数
        properties.setProperty(ProducerConfig.RETRIES_CONFIG,"5");
        //指定失败重试的间隔时间
        properties.setProperty(ProducerConfig.RETRY_BACKOFF_MS_CONFIG,"3000");
        //指定消息的压缩方式
        properties.setProperty(ProducerConfig.COMPRESSION_TYPE_CONFIG,"gzip");


        //2、创建生产者
        KafkaProducer<String,String> producer = new KafkaProducer<>(properties);

        //3、准备消息 hash("key001")
        //ProducerRecord<String, String> recored = new ProducerRecord<>("topic001","value001");


       
        //4、发送消息
        //方式1：单边发送，只管发，不管有没有发成功
        for (int i = 0; i < 5; i++) {

            if(i%2==0){
                ProducerRecord<String, String> recored = new ProducerRecord<>("topicA",null,"hello kafka");
                producer.send(recored);
            }else{
                ProducerRecord<String, String> recored = new ProducerRecord<>("topicA",null,"hello itcast");
                producer.send(recored);
            }

        }

        
        //方式2：同步发送，发完之后还要确定有没有发成功，如果没有发面功，会自动重试.安全性最高，容易形成阻塞
//        for (int i = 0; i <100; i++) {
//
//            ProducerRecord<String, String> recored = new ProducerRecord<>("topic001","value001"+i);
//            try {
//                RecordMetadata recordMetadata = producer.send(recored).get();
//            } catch (InterruptedException e) {
//                throw new RuntimeException(e);
//            } catch (ExecutionException e) {
//                throw new RuntimeException(e);
//            }
//        }


        //方式3：异步发送
//        for (int i = 0; i <100; i++) {
//
//            ProducerRecord<String, String> recored = new ProducerRecord<>("topic001","value001"+i);
//            producer.send(recored, new Callback() {
//                @Override
//                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
//                    if(e!=null){
//                        System.out.println("发到"+recordMetadata.offset()+"位置");
//                        e.printStackTrace();
//                    }
//                }
//            });
//        }

        //5、关闭producer
        producer.close();
    }
}
